package day07

import day07.caseClass.CategoryCountInfo
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * 6.2	需求1：Top10热门品类
 *
 * 需求说明：品类是指产品的分类，大型电商网站品类分多级，咱们的项目中品类只有一级，不同的公司可能对热门的定义不一样。
 * 我们按照每个品类的点击、下单、支付的量来统计热门品类。
 *
 * 鞋			  点击数 下单数  支付数
 * 衣服		  点击数 下单数  支付数
 * 生活用品	点击数 下单数  支付数
 *
 * 例如，综合排名 = 点击数 * 20% + 下单数 * 30% + 支付数 * 50%
 *
 * 本项目需求优化为：先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
 */
object Spark01_TopN_Req1 {

}

/**
 *
 * 6.2.1	代码实现方式一
 */
object Spark01_TopN_Req1_1 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val categoryCountInfoRdd: RDD[(String, CategoryCountInfo)] = TopNUtils.getCategoryCountInfoRdd(sc)

    // 根据品类id进行分组
    // Tips：groupByKey：按照key进行分组，直接进行shuffle
    val categoryCountInfoMap: RDD[(String, Iterable[CategoryCountInfo])] = categoryCountInfoRdd.groupByKey

    // 对分组后的value进行聚合，并将结果进行转换，取CategoryCountInfo
    val mapRdd: RDD[CategoryCountInfo] = categoryCountInfoMap.mapValues {
      categories: Iterable[CategoryCountInfo] => {
        val info: CategoryCountInfo = categories.reduce((category1, category2) => {
          category1.clickCount = category1.clickCount + category2.clickCount
          category1.orderCount = category1.orderCount + category2.orderCount
          category1.payCount = category1.payCount + category2.payCount
          category1
        })
        info
      }
    }.map(_._2)

    // 按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
    val top10: Array[CategoryCountInfo] = mapRdd.sortBy(category => (category.clickCount, category.orderCount, category.payCount), ascending = false).take(num = 10)

    top10.foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 6.2.2	代码实现方式二：使用reduceByKey的预聚合优化
 */
object Spark01_TopN_Req1_2 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val categoryCountInfoRdd: RDD[(String, CategoryCountInfo)] = TopNUtils.getCategoryCountInfoRdd(sc)

    // 根据品类id进行预聚合，并将结果进行转换，取CategoryCountInfo
    // Tips： reduceByKey：按照Key进行聚合，在shuffle之前有combine(预聚合)操作，返回结果是RDD[k,v]
    val mapRdd: RDD[CategoryCountInfo] = categoryCountInfoRdd.reduceByKey((category1, category2) => {
      category1.clickCount = category1.clickCount + category2.clickCount
      category1.orderCount = category1.orderCount + category2.orderCount
      category1.payCount = category1.payCount + category2.payCount
      category1
    }).map(_._2)

    // 按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
    val top10: Array[CategoryCountInfo] = mapRdd.sortBy(category => (category.clickCount, category.orderCount, category.payCount), ascending = false).take(num = 10)

    top10.foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 6.2.3	代码实现方式三：采用累加器，避免shuffle过程
 *
 * 思路2
 * 最好的办法应该是遍历一次能够计算出来上述的3个指标。使用累加器可以达成我们的需求。
 * （1）遍历全部日志数据，根据品类id和操作类型分别累加，需要用到累加器
 * 定义累加器，当碰到订单和支付业务的时候注意拆分字段才能得到品类 id
 * （2）遍历完成之后就得到每个品类 id 和操作类型的数量.
 * （3）按照点击下单支付的顺序来排序
 * （4）取出 Top10
 *
 */
object Spark01_TopN_Req1_3 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val topNList: List[CategoryCountInfo] = TopNUtils.getTopNList(sc)
    topNList.foreach(println)

    //4.关闭连接
    sc.stop()
  }
}
